package org.duang.proxy;

import java.io.UnsupportedEncodingException;
import java.lang.reflect.Method;

import org.duang.aop.Proxy;
import org.duang.aop.ProxyChain;
import org.duang.common.exceptios.ServiceException;
import org.duang.config.Config;
import org.duang.ioc.Mq;
import org.duang.kit.ToolsKit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun4j.core.ONSUtils;
import com.aliyun4j.listener.ONSListener;
import com.aliyun4j.ons.ONSEnum;
import com.aliyun4j.ons.ONSMessage;
import com.aliyun4j.utils.Reflections;

/**
 * 用于ONS队列
 * @author laotang
 */
public class MqProxy implements Proxy {
	
	private static final Logger logger = LoggerFactory.getLogger(MqProxy.class);
	
	private Object resultObj;
	
	public Object doProxy(final ProxyChain proxyChain) throws Exception {
		long startTime = System.currentTimeMillis();
		resultObj = null;
		Method method = proxyChain.getTargetMethod();
		final Mq mq = method.getAnnotation(Mq.class);
		if(null != mq){
			ONSEnum enums = mq.value();
			if(ToolsKit.isEmpty(enums.getId())) {
					throw new NullPointerException("consumerId is empty!");
			}
			try {
				ONSUtils.getMessage(enums, new ONSListener(){
					public Action consume(Message message, ConsumeContext context) {
						try {
							String body = new String(message.getBody(), Config.getEncoding());
							ONSMessage onsMessage = ToolsKit.jsonParseObject(body, ONSMessage.class);
							String jsonText = ToolsKit.toJsonString(onsMessage.getBody());
							onsMessage.setBody(jsonText);
							Object[] params = {onsMessage};
							//TODO ...处理重复
							resultObj = proxyChain.doProxyChain(params);
						} catch(UnsupportedEncodingException uee){
							logger.warn("ONS byte Convert String Fail: " + uee.getMessage(), uee);
							return Action.ReconsumeLater;			// 重投
						}catch (ServiceException se){
							logger.warn("ONS byte Convert String Fail: " + se.getMessage(), se);
							return Action.ReconsumeLater;		// 重投
						} catch (Exception e) {
							logger.warn("MqProxy doProxy is Fail: " + e.getMessage(), e);
							return Action.ReconsumeLater;		// 重投
						}
						return Action.CommitMessage;
					}
				});
			} catch (Exception e) {
				logger.warn("ONS Consumer getMessage Fail: " + e.getMessage(), e);
			}
		} else {
			resultObj = proxyChain.doProxyChain();
		}
//		if((System.currentTimeMillis() - startTime) >= 300) {
			Class<?> cls = proxyChain.getTargetClass();
			logger.warn("ONS Consumer run time glt 300: " + cls.getName() +"=====>"+ method.getName());
//		}
		return resultObj;
	}
}
